package cn;

import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.CharBuffer;

import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Context;
import org.zeromq.ZMQ.Socket;

/*
/usr/local/lib  
# ls *zmq*
libjzmq.a   libjzmq.so    libjzmq.so.0.0.0  libzmq.la  libzmq.so.3
libjzmq.la  libjzmq.so.0  libzmq.a          libzmq.so  libzmq.so.3.0.0
[root@localhost lib]# cp *zmq* /usr/lib64/. 
 */
public class jzmqDemo {

	public static void main(String[] args) {
		jzmqDemo m = new jzmqDemo();
		if (args.length >= 1) {
			m.runJNI();
		} else {
			m.runJNINative();
		}
	}

	private Context context;
	private Socket socket;
	int ITEMS = 25;
	int QUEUE_SIZE = 10000;

	private void runJNI() {
		// Logger.debug("启动ZMQ接收进程...");
		this.context = ZMQ.context(1);
		this.socket = this.context.socket(ZMQ.PULL);
		this.socket.bind("tcp://127.0.0.1:7788");
		this.socket.setHWM(QUEUE_SIZE);
		// this.socket.setLinger(0);
		// Logger.debug("ZMQ接收进程启动成功，监听端口： " + Globals.RECV_DATA_PORT);

		ByteBuffer buffer = ByteBuffer.allocate(10240).order(ByteOrder.nativeOrder());
		System.out.println("start runJNI....");
		int count = 0;
		byte[] len = new byte[4];

		while (true) {
			for (int k = 0; k < ITEMS; k++) {
				len = this.socket.recv(0);
				String s = this.socket.recvStr();
				// System.out.println(s);

				len = this.socket.recv(0);
				s = this.socket.recvStr();
				// System.out.println(s);
			}

			// len = this.socket.recv(0);
			buffer.position(0);
			this.socket.recv(buffer.array(), 0, 4, 0);
			buffer.position(0);

			System.out.println("get:" + count++ + "==" + buffer.getInt());
			// System.out.println("");
		}

	}

	private void runJNINative() {
		// Logger.debug("启动ZMQ接收进程...");
		this.context = ZMQ.context(1);
		this.socket = this.context.socket(ZMQ.PULL);
		this.socket.bind("tcp://127.0.0.1:7788");
		this.socket.setHWM(QUEUE_SIZE);
		// this.socket.setLinger(0);
		// Logger.debug("ZMQ接收进程启动成功，监听端口： " + Globals.RECV_DATA_PORT);

		// ByteBuffer buffer =
		// ByteBuffer.allocateDirect(1024).order(ByteOrder.nativeOrder());
		ByteBuffer buffer = ByteBuffer.allocateDirect(10240).order(ByteOrder.nativeOrder());

		Runnable t = () -> {
			buffer.position(0);
			CharBuffer cc = buffer.asCharBuffer();
			for (int c = 0; c < 4; c++) {
				System.out.print((int) cc.charAt(c));
				System.out.println();
			}
		};

		System.out.println("start runJNINative....");
		byte[] bs = new byte[10240];
		int l = 0;
		int count = 0;
		while (true) {
			buffer.position(0);

			// String s = this.socket.recvStr();
			String str = "";
			for (int k = 0; k < ITEMS; k++) {
				// int last = buffer.position();
				buffer.position(0);
				// bs= this.socket.recv();
				this.socket.recvZeroCopy(buffer, 4, 0);
				buffer.position(0);
				// CharBuffer cc = buffer.asCharBuffer();
				// for(int c=0; c< 4; c++){
				// System.out.print((int)cc.charAt(c));
				// System.out.println();
				// }
				int len = buffer.getInt();
				// System.out.println("first len:" + len);
				this.socket.recvZeroCopy(buffer, len, 0);
				// t.run();

				buffer.position(0);
				buffer.get(bs, 0, len);
				str = new String(bs, 0, len);
				// System.out.println(str);

				// last = buffer.position();
				this.socket.recvZeroCopy(buffer, 4, 0);
				buffer.position(0);
				len = buffer.getInt();
				// System.out.println("first len:" + len);

				buffer.position(0);
				this.socket.recvZeroCopy(buffer, len, 0);
				// t.run();
				buffer.position(0);
				buffer.get(bs, 0, len);
				str = new String(bs, 0, len);
				// System.out.println(str);
			}

			// System.out.println(str);
			// this.socket.recvByteBuffer(buffer, 0);
			buffer.position(0);
			this.socket.recvZeroCopy(buffer, 4, 0);
			// bs= this.socket.recv();
			int readc = buffer.position();
			// System.out.println(readc);
			buffer.position(0);
			l = buffer.getInt();
			// System.out.println("first len:" + l);

			// System.out.println(s);
			// System.out.println(l);
			System.out.println("get:" + count++);
			// System.out.println("");
		}
	}

}
